package com.thoughtful.remoting.transport.netty.codec;

import java.util.Arrays;

import com.thoughtful.compress.Compress;
import com.thoughtful.remoting.constants.RpcConstants;
import com.thoughtful.remoting.dto.RpcMessage;
import com.thoughtful.remoting.dto.RpcRequest;
import com.thoughtful.remoting.dto.RpcResponse;
import com.thoughtful.serialize.Serializer;
import com.thougtful.enums.CompressTypeEnum;
import com.thougtful.enums.SerializationTypeEnum;
import com.thougtful.extension.ExtensionLoader;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;

/**
 * <p>
 * 自定义协议解码器
 * <pre>
 *   0     1     2     3     4        5     6     7     8         9          10      11     12  13  14   15 16
 *   +-----+-----+-----+-----+--------+----+----+----+------+-----------+-------+----- --+-----+-----+-------+
 *   |   magic   code        |version | full length         | messageType| codec|compress|    RequestId       |
 *   +-----------------------+--------+---------------------+-----------+-----------+-----------+------------+
 *   |                                                                                                       |
 *   |                                         body                                                          |
 *   |                                                                                                       |
 *   |                                        ... ...                                                        |
 *   +-------------------------------------------------------------------------------------------------------+
 * 4B magic code (魔法数)    1B version (版本)       4B full length (消息长度)    1B messageType (消息类型)
 * 1B codec (序列化类型)     1B compress (压缩类型)   4B  requestId (请求的Id)
 * body (object类型数据)
 * </pre>
 *
 * <p>
 *     {@link LengthFieldBasedFrameDecoder} 是一个基于长度解码器, 用于解决 TCP 的拆包和粘包问题
 * </p>
 *
 *
 * @author yushun
 * @createTime 2022/10/18 - 10:31
 * @see <a href="https://zhuanlan.zhihu.com/p/95621344">LengthFieldBasedFrameDecoder解码器</a>
 **/
public class RpcMessageDecoder extends LengthFieldBasedFrameDecoder {

    public RpcMessageDecoder() {
        // lengthFieldOffset:   因为魔法数占用 4B, 版本占用 1B, 需要跳过这两个字段才是长度字段, 所以设置为 5
        // lengthFieldLength:   长度字段规定是 4B, 所以为 4
        // lengthAdjustment:    lengthFieldLength 记录的完整长度包括所有的数据, 需要减去前面已经读到的 9 个字节, 所以为 -9
        // initialBytesToStrip: 我们回去手动检查魔法代码和版本, 所以不用删除任何字节, 所以设置为 0
        this(RpcConstants.MAX_FRAME_LENGTH, 5, 4, -9, 0);
    }

    /**
     * @param maxFrameLength        最大帧长度, 它决定可以接收的最大数据长度. 如果超过, 数据将被丢弃
     * @param lengthFieldOffset     长度字段的偏移量, 表示需要在数据帧前面跳过多少字节才是长度字段
     * @param lengthFieldLength     长度字段本身的长度
     * @param lengthAdjustment      要添加到长度字段值中的补偿值
     * @param initialBytesToStrip   跳过的字节数. 如果需要接收请求头和请求体的数据, 就设置为0, 如果只要接收请求体数据, 则设置为需要跳过的请求头占用的字节数
     */
    public RpcMessageDecoder(int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip) {
        super(maxFrameLength, lengthFieldOffset, lengthFieldLength, lengthAdjustment, initialBytesToStrip);
    }

    @Override
    protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        Object decoded = super.decode(ctx, in);
        if (decoded instanceof ByteBuf) {
            ByteBuf frame = (ByteBuf)decoded;
            if (frame.readableBytes() >= RpcConstants.TOTAL_LENGTH) {
                try {
                    return decodeFrame(frame);
                } catch (Exception e) {
                    throw new RuntimeException("Decode frame error! ", e);
                } finally {
                    frame.release();
                }
            }
        }
        return decoded;
    }

    private Object decodeFrame(ByteBuf frame) {
        // 注意: 需要按照顺序读取
        checkMagicNumber(frame);
        checkVersion(frame);
        int fullLength = frame.readInt();
        // 构建 RpcMessage Object
        byte messageType = frame.readByte();
        byte codecType = frame.readByte();
        byte compressType = frame.readByte();
        int requestId = frame.readInt();
        RpcMessage rpcMessage = RpcMessage.builder()
            .codec(codecType)
            .requestId(requestId)
            .compress(compressType)
            .messageType(messageType)
            .build();

        if (messageType == RpcConstants.HEARTBEAT_REQUEST_TYPE) {
            rpcMessage.setData(RpcConstants.PINT);
            return rpcMessage;
        }
        if (messageType == RpcConstants.HEARTBEAT_RESPONSE_TYPE) {
            rpcMessage.setData(RpcConstants.PONG);
            return rpcMessage;
        }

        int bodyLength = fullLength - RpcConstants.HEAD_LENGTH;
        if (bodyLength > 0) {
            byte[] bs = new byte[bodyLength];
            frame.readBytes(bs);

            // 解压
            String compressName = CompressTypeEnum.getName(compressType);
            Compress compress = ExtensionLoader.getExtensionLoader(Compress.class).getExtension(compressName);
            bs = compress.decompress(bs);

            // 反序列化
            String codecName = SerializationTypeEnum.getName(codecType);
            Serializer serializer = ExtensionLoader.getExtensionLoader(Serializer.class).getExtension(codecName);
            if (messageType == RpcConstants.REQUEST_TYPE) {
                RpcRequest rpcRequest = serializer.deserializer(bs, RpcRequest.class);
                rpcMessage.setData(rpcRequest);
            } else if (messageType == RpcConstants.RESPONSE_TYPE) {
                @SuppressWarnings("rawtypes")
                RpcResponse rpcRequest = serializer.deserializer(bs, RpcResponse.class);
                rpcMessage.setData(rpcRequest);
            } else {
                rpcMessage.setData(bs);
            }
        }
        return rpcMessage;
    }

    private void checkVersion(ByteBuf frame) {
        byte version = frame.readByte();
        if (version != RpcConstants.VERSION) {
            throw new IllegalArgumentException("version isn't compatible, version: " + version);
        }
    }

    private void checkMagicNumber(ByteBuf frame) {
        int len = RpcConstants.MAGIC_NUMBER.length;
        byte[] tmp = new byte[len];
        frame.readBytes(tmp);
        for (int i = 0; i < len; i++) {
            if (tmp[i] != RpcConstants.MAGIC_NUMBER[i]) {
                throw new IllegalArgumentException("Unknown magic code: " + Arrays.toString(tmp));
            }
        }
    }
}
